package com.shujia.flink.sink

import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.client.{HConnection, HConnectionManager, HTableInterface, Put}


class HbaseSink[IN](tableName: String, hbaseMapper: HbaseMapper[IN], zkQuorum: String) extends RichSinkFunction[IN] {

  var connection: HConnection = _

  var table: HTableInterface = _

  //open方法每一个task只执行一次，  一般用户打开外部连接
  override def open(parameters: Configuration): Unit = {
    //创建hadoop配置文件对象
    //在类名冲突的时候将其中一个改成类全名
    val configuration = new org.apache.hadoop.conf.Configuration

    //指定zk地址
    configuration.set("hbase.zookeeper.quorum", zkQuorum)
    //建立连接   和zk的连接
    connection = HConnectionManager.createConnection(configuration)

    table = connection.getTable(tableName)
  }

  //关闭连接
  override def close(): Unit = {

    if (connection != null) {
      connection.close()
    }
  }

  //每一条数据执行一次
  override def invoke(value: IN, context: SinkFunction.Context[_]): Unit = {
    val put = hbaseMapper.mapper(value)
    table.put(put)
  }
}
